package com.gizwits.RabbitMQReceiver

import com.gizwits.util.RabbitMQUtils
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming._
import org.apache.spark.{SparkConf, SparkContext}
import reactivemongo.api._
import reactivemongo.api.collections.bson.BSONCollection

import reactivemongo.bson._
object RabbitMQReceiver {
  def main(args: Array[String]) {

    //val Array(zkQuorum, group, topics, numThreads) = args
  //  val Array(rabbitMQHost, rabbitMQPort, rabbitMQQueueName)=args
    val sparkConf = new SparkConf().setAppName("RaddbitHelloWorld")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(2))
    ssc.checkpoint(".")
    val rabbitMQHosts="s5"
    val rbs = RabbitMQUtils.createStreamFromRoutingKeys(ssc,Option("topqueue"),
      rabbitMQHosts,
      5672,
      "topic_logs_exchange",
      List("log.*"),
      Option("topic"),
      false,
      StorageLevel.MEMORY_AND_DISK_SER_2)


    val driver = new MongoDriver
    val connection = driver.connection(List("s5:27017"))


   // val db = connection("adlogdb")
    //val collection = db[BSONCollection]("test")
   // Macros.handler.
    //  rbs.foreachRDD(line=>)
    //
   // rbs.print()
    ssc.start()
    //
     Thread.sleep(3000)
     ssc.awaitTermination()

    //val Array(rabbitMQHost, rabbitMQPort, rabbitMQQueueName, storageLevel)=("s5",15672,"gizwits_rt_data",StorageLevel.MEMORY_AND_DISK_2)

    //val  rabbitMQHost=15672
    //val  rabbitMQQueueName="gizwits_rt_data"
   // val storageLevel=StorageLevel.MEMORY_AND_DISK_2



   // val rbs=RabbitMQUtils.createStreamFromAQueue(ssc, rabbitMQHost,5672,rabbitMQQueueName,false)
  // val  word= rbs.map(line=>line).window(Minutes(10), Seconds(2))

   //  word.print()
   // val words = rbs.map(_._2).flatMap(_.split(" "))
  //  val pairs = words.map(word => (word, 1))
    // val wordCounts = pairs.reduceByKey(_ + _)
  //  val wordCounts = words.map(x => (x, 1L))
    //  .reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2)
   // wordCounts.print()


//   val stateDstream = pairs.updateStateByKey((values: Seq[Int], state: Option[Int]) => {
//      val currentCount = values.sum
//
//      val previousCount = state.getOrElse(0)
//
//      Some(currentCount + previousCount)
//    })
//    stateDstream.print()

   // Thread.sleep(3000)
   // rbs.stop()
  }
}
